package com.atguigu.flink03;

import com.atguigu.beans.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Felix
 * @date 2024/2/20
 * 该案例演示了规约算子 reduce
 * 需求：通过reduce算子，实现max效果
 * reduce算子细节
 *      当第一条数据到来的时候，reduce方法不会执行，直接将数据向下输出
 *      reduce(参数1，参数2)
 *      参数1：规约的结果
 *      参数2：流中新来的数据
 */
public class Flink12_Trans_Reduce {
    public static void main(String[] args) throws Exception {
        //TODO 1.指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //TODO 2.从指定的网络端口读取水位数据  ws1,1,10
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 3.对读取的数据类型进行转换   String->WaterSensor
        SingleOutputStreamOperator<WaterSensor> wsDS = socketDS.map(
                new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String lineStr) throws Exception {
                        String[] strArr = lineStr.split(",");
                        return new WaterSensor(strArr[0], Long.parseLong(strArr[1]), Integer.parseInt(strArr[2]));
                    }
                }
        );
        //TODO 4.按照传感器id进行分组
        KeyedStream<WaterSensor, String> keyedDS = wsDS.keyBy(WaterSensor::getId);

        //TODO 5.规约聚合
        SingleOutputStreamOperator<WaterSensor> reduceDS = keyedDS.reduce(
                new ReduceFunction<WaterSensor>() {
                    @Override
                    public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                        System.out.println("value1:" + value1);
                        System.out.println("value2:" + value2);
                        int max = Math.max(value1.vc, value2.vc);
                        value1.setVc(max);
                        return value1;
                    }
                }
        );
        reduceDS.print();

        env.execute();

    }
}
